package com.xdu.yunwufog.connection.netty;

import com.google.gson.*;
import com.xdu.yunwufog.connection.utils.DataBusConstant;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;



import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;


public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 使用阻塞式LinkedBlockingQueue，对应响应结果保存,并发安全，响应结果为String
     * 用于记录通道响应的结果集合
     */
    private static final Map<Long, LinkedBlockingDeque<String>> RESULT_MAP = new ConcurrentHashMap<>();
    volatile static Map<Integer, Set<Channel>> coreChannel = new HashMap<>();
    private final Logger logger = LoggerFactory.getLogger(getClass());


    public String sendMessage(ByteBuf message, Channel ch){
        LinkedBlockingDeque<String> linked = new LinkedBlockingDeque<>();
        // 获取channel中存储的全局唯一随机值
        Long randomId = ch.attr(AttributeKey.<Long>valueOf(DataBusConstant.RANDOM_KEY)).get();
        RESULT_MAP.put(randomId,linked);
        // 发送message
        ch.writeAndFlush(message);
        String res = null;
        try {
            // 设置100s的获取超时时间或使用take()---获取不到返回结果则一直阻塞
            res = RESULT_MAP.get(randomId).poll(3, TimeUnit.MINUTES);
            RESULT_MAP.remove(randomId);
        }catch (Exception e){
            e.printStackTrace();
        } finally {
            if(res == null){
                res = "{'code':'500'}";
            }
        }
        return res;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        logger.debug("[DEBUG] into channelRead");
        String message = null;
        if (msg instanceof String){
            message = msg.toString();
        }else if (msg instanceof ByteBuf){
            message = ((ByteBuf)msg).toString(CharsetUtil.UTF_8);
        }
        if(message == null){
            logger.warn("[WARN] Receive null message!");
            return;
        }

        // 获取channel中存储的全局唯一随机值
        Long randomId = ctx.channel().attr(AttributeKey.<Long>valueOf(DataBusConstant.RANDOM_KEY)).get();

        logger.info("[INFO] Server return message length:{}", message.length());
        // 将服务端返回结果返回对应的channel中
        LinkedBlockingDeque<String> linked = RESULT_MAP.get(randomId);
        Map<String,String> map = new HashMap<>();
        map.put("code","200");
        map.put("data",message);
        linked.add(new Gson().toJson(map));

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        boolean active = ctx.channel().isActive();
        logger.debug("[DEBUG] Now,channel status is {}",active);
    }

    /**
     * 心跳机制实现连接的动态回收
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        logger.info("[INFO] [Client heartbeat monitoring sending...] Channel Num:{}",ctx.channel().id());
        Channel channel = ctx.channel();
        if (evt instanceof IdleStateEvent){
            // 当客户端开始发送心跳检测时，说明没有业务请求，释放通道数设定的CORE_CONNECTIONS
            if (channel.isActive()){
                // 使用pool的hash作为key，维护CORE_CONNECTIONS个通道数，多余关闭
                int poolHash = NettyClientPool.getPoolHash(channel);
                // 获取poolHash对应的连接集合
                Set<Channel> channels = coreChannel.get(poolHash);
                channels = channels == null ? new HashSet<>(DataBusConstant.CORE_CONNECTIONS) : channels;
                channels.add(channel);
                if (channels.stream().filter(Channel::isActive).count() > DataBusConstant.CORE_CONNECTIONS){
                    logger.info("[INFO] Close exist channel beyond CORE_CONNECTIONS: {}",channel.id());
                    channels.remove(channel);
                    channel.close();
                }
                // 将更新后的连接集合到coreChannel中
                coreChannel.put(poolHash,channels);
            }
        } else {
            super.userEventTriggered(ctx,evt);
        }
    }
}
